package com.atguigu.tingshu.common.service;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;


    /**
     * 发送kafka消息
     * @param topic 话题名称
     * @param data 业务数据
     */
    public void sendMessage(String topic, String data) {
        this.sendMessage(topic, null, data);
    }


    /**
     * 生产者发送kafka消息
     *
     * @param topic 话题名称
     * @param key   数据键
     * @param data  数据值
     */
    public void sendMessage(String topic, String key, String data) {
        //1.发送普通消息
        CompletableFuture future = kafkaTemplate.send(topic, key, data);
        //2.判断发送消息结果
        future.whenComplete((metaData, exception) -> {
            //metaData(SendResult)发送后元数据
            if (exception != null) {
                logger.error("[Kafka生产者]发送消息异常：{}", exception);
            } else {
                SendResult<String, String> sendResult = (SendResult<String, String>) metaData;
                RecordMetadata recordMetadata = sendResult.getRecordMetadata();
                logger.info("[kafka生产者]发送消息成功,话题：{}，分区：{}，偏移量：{}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
            }
        });
    }
}
